package com.example.study;


import java.util.concurrent.TimeUnit;

import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.schedulers.Schedulers;

public class RxJavaLearn {

    public static void main(String[] args) {
//        method_1();
//        method_2();
        method_3();
        try {
            Thread.sleep(20000);//主线程休眠
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 定时任务
     * 关键字符 timer
     */
    private static void method_1() {
        Observable.timer(2000, TimeUnit.MILLISECONDS)//延时2000毫秒执行
                .subscribeOn(Schedulers.io())//上游事件在子线程中处理
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {
                        printlnString.printlnString("开始订阅");
                    }

                    @Override
                    public void onNext(@NonNull Long aLong) {
                        printlnString.printlnString("onNext =" + aLong);
                    }

                    @Override
                    public void onError(@NonNull Throwable e) {
                        printlnString.printlnString("onError = " + "onError");
                    }

                    @Override
                    public void onComplete() {
                        printlnString.printlnString("完成");
                    }
                });
    }

    /**
     * 定时任务
     * 关键字符interval
     */
    static Disposable mDisposable;

    private static void method_2() {

        Observable.interval(2000, 1000, TimeUnit.MILLISECONDS)//延迟2000毫秒每隔2000执行一直任务
                .subscribeOn(Schedulers.io())// 约定观察者在在子线程中执行 io操作比较多的
                .observeOn(Schedulers.io())//定时任务不能在主线程中
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {
                        printlnString.printlnString("开始订阅");
                        mDisposable = d;//保存mDisposable
                    }

                    @Override
                    public void onNext(@NonNull Long aLong) {
                        printlnString.printlnString("onNext =" + aLong);
                        if (aLong == 10) {
                            onComplete();
                            if (mDisposable != null && !mDisposable.isDisposed())
                                mDisposable.dispose();//取消订阅
                        }
                    }

                    @Override
                    public void onError(@NonNull Throwable e) {
                        printlnString.printlnString("onError = " + "onError");
                    }

                    @Override
                    public void onComplete() {
                        printlnString.printlnString("完成");
                    }
                });

    }

    /**
     * 批量处理事件
     */

    private static void method_3() {
        Observable.just("name1", "name2", "name3")
                .subscribeOn(Schedulers.computation())//CPU计算类型的线程
//                .observeOn(AndroidSchedulers.mainThread())//android的
                .observeOn(Schedulers.io())// 子线程中
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {
                        printlnString.printlnString("开始订阅");
                    }

                    @Override
                    public void onNext(@NonNull String s) {
                        printlnString.printlnString("onNext =" + s + " time =" + System.currentTimeMillis());
                    }

                    @Override
                    public void onError(@NonNull Throwable e) {
                        printlnString.printlnString("onError = " + "onError");
                    }

                    @Override
                    public void onComplete() {
                        printlnString.printlnString("完成");
                    }
                });
    }


}
